package com.gitee.melin.bee.core.jdbc.dialect.impl;

import com.gitee.melin.bee.core.jdbc.dialect.AbstractJdbcDialect;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.relational.BeeTableType;
import com.gitee.melin.bee.core.jdbc.relational.ConnectionInfo;
import com.gitee.melin.bee.core.jdbc.relational.MetaColumn;
import com.google.common.collect.Lists;
import java.sql.*;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PostgreSqlJdbcDialect extends AbstractJdbcDialect {

    private static final Logger LOG = LoggerFactory.getLogger(PostgreSqlJdbcDialect.class);

    protected static final List<String> EXCLUDED_SCHEMAS = Collections.unmodifiableList(Arrays.asList(
            "information_schema",
            "pg_catalog",
            "root",
            "pg_toast",
            "pg_temp_1",
            "pg_toast_temp_1",
            "postgres",
            "template0",
            "template1"));

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.POSTGRESQL;
    }

    private static final String QUERY_DATABASES_SQL =
            "select datname from pg_catalog.pg_database where datistemplate = false";

    private static final String VERSION_SQL = "SELECT version()";

    private static final double SUPPORT_RELPARTBOUND_START_VERSION = 10;

    private static final String SUPPORT_RELPARTBOUND_STATEMENT = "AND c.relpartbound IS NULL";

    private static final String NOT_SUPPORT_RELPARTBOUND_STATEMENT =
            "and not EXISTS(SELECT 1 from pg_inherits  where pg_inherits.inhrelid = c.oid)";

    private Integer isHighVersion = 0; // 高版本支持relpartbound字段:1,低于10下版本不支持 -1

    public PostgreSqlJdbcDialect(ConnectionInfo connectionInfo) {
        super(connectionInfo);
    }

    public PostgreSqlJdbcDialect(Connection connection) {
        super(connection);
    }

    private void sqlCompatibleVersion() {
        if (isHighVersion != 0) {
            return;
        }

        try (Connection connection = this.getConnection();
                Statement statement = connection.createStatement();
                ResultSet versionRs = statement.executeQuery(VERSION_SQL)) {
            String version = null;
            while (versionRs.next()) {
                version = versionRs.getString(1);
            }

            if (version.contains("(")) {
                version = version.substring(0, version.indexOf("("));
                version = version.replaceFirst("PostgreSQL", "").trim();
            }

            if (version.contains(".")) {
                version = version.split("\\.")[0];
            }
            double versionNumber = Double.parseDouble(version);
            if (versionNumber > SUPPORT_RELPARTBOUND_START_VERSION) {
                isHighVersion = 1;
            } else {
                isHighVersion = -1;
            }
        } catch (Exception ex) {
            isHighVersion = 1;
            LOG.error(ex.getMessage(), ex);
        }
    }

    @Override
    protected Driver getDriver() throws SQLException {
        return loadDriver(DataSourceType.POSTGRESQL.getDriverClass());
    }

    @Override
    protected boolean filterSchema(String schemaName) {
        return !EXCLUDED_SCHEMAS.contains(schemaName);
    }

    @Override
    protected String getDDLColumnName(BeeTableType type) {
        return "ct_script";
    }

    @Override
    public List<String> getDatabases() {
        try (Connection connection = this.getConnection();
                PreparedStatement preparedStatement = connection.prepareStatement(QUERY_DATABASES_SQL);
                ResultSet resultSet = preparedStatement.executeQuery()) {

            List<String> catalogs = Lists.newArrayList();
            while (resultSet.next()) {
                String catalogName = resultSet.getString(1);
                catalogs.add(catalogName);
            }
            return catalogs;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    protected String getSelectTablesSql(String tableName) {
        String sql = "SELECT n.nspname              AS TABLE_SCHEMA\n"
                + "     , c.relname              AS TABLE_NAME\n"
                + "     , c.relkind as TABLE_TYPE\n"
                + "     , null      AS TABLE_ENGINE\n"
                + "     , obj_description(c.oid) AS TABLE_COMMENT\n"
                + "     , c.reltuples AS TABLE_ROWS\n"
                + "     , cast(c.relpages as bigint) * 8 * 1024 as DATA_SIZE\n"
                + "     , (CASE WHEN c.relkind = 'p' THEN 'p' ELSE null END) as TABLE_PARTITION\n"
                + "     , null AS CREATE_TIME\n"
                + "     , null AS UPDATE_TIME\n"
                + "FROM pg_class c\n"
                + "         LEFT JOIN pg_namespace n\n"
                + "                   ON n.oid = c.relnamespace\n"
                + "WHERE c.relkind in ('r', 'f', 'p', 'v', 'm')\n"
                + "  AND n.nspname = :schemaName";

        if (StringUtils.isNotBlank(tableName)) {
            sql += " AND c.relname = :tableName";
        }
        return sql;
    }

    @Override
    protected String getSelectColumnsSql() {
        return "SELECT\n"
                + "    a.attname AS column_name,\n"
                + "    t.typname AS DATA_TYPE,\n"
                + "    CASE\n"
                + "        WHEN a.atttypmod = -1 THEN t.typname\n"
                + "        WHEN t.typname = 'varchar' THEN t.typname || '(' || (a.atttypmod - 4) || ')'\n"
                + "        WHEN t.typname = 'bpchar' THEN 'char' || '(' || (a.atttypmod - 4) || ')'\n"
                + "        WHEN t.typname = 'numeric' OR t.typname = 'decimal' THEN t.typname || '(' || ((a.atttypmod - 4) >> 16) || ', ' || ((a.atttypmod - 4) & 65535) || ')'\n"
                + "        WHEN t.typname = 'bit' OR t.typname = 'bit varying' THEN t.typname || '(' || (a.atttypmod - 4) || ')'\n"
                + "        WHEN t.typname IN ('time', 'timetz', 'timestamp', 'timestamptz') THEN t.typname || '(' || a.atttypmod || ')'\n"
                + "        ELSE t.typname || ''\n"
                + "    END AS COLUMN_TYPE,\n"
                + "    CASE\n"
                + "        WHEN a.atttypmod = -1 THEN NULL\n"
                + "        WHEN t.typname IN ('varchar', 'bpchar', 'bit', 'bit varying') THEN a.atttypmod - 4\n"
                + "        WHEN t.typname IN ('numeric', 'decimal') THEN (a.atttypmod - 4) >> 16\n"
                + "        ELSE NULL\n"
                + "    END AS DATA_LENGTH,\n"
                + "    CASE\n"
                + "        WHEN a.atttypmod = -1 THEN NULL\n"
                + "        WHEN t.typname IN ('numeric', 'decimal') THEN (a.atttypmod - 4) & 65535\n"
                + "        WHEN t.typname IN ('time', 'timetz', 'timestamp', 'timestamptz') THEN a.atttypmod\n"
                + "        ELSE NULL\n"
                + "    END AS NUMERIC_SCALE,\n"
                + "    NULL as NUMERIC_PRECISION,\n"
                + "    null AS DATETIME_PRECISION,\n"
                + "    d.description AS COLUMN_COMMENT,\n"
                + "    pg_get_expr(ad.adbin, ad.adrelid) AS DEFAULT_VALUE,\n"
                + "    CASE WHEN a.attnotnull THEN 'NO' ELSE 'YES' END AS IS_NULLABLE\n"
                + "FROM\n"
                + "    pg_class c\n"
                + "    JOIN pg_namespace n ON c.relnamespace = n.oid\n"
                + "    JOIN pg_attribute a ON a.attrelid = c.oid\n"
                + "    JOIN pg_type t ON a.atttypid = t.oid\n"
                + "    LEFT JOIN pg_description d ON c.oid = d.objoid AND a.attnum = d.objsubid\n"
                + "    LEFT JOIN pg_attrdef ad ON a.attnum = ad.adnum AND a.attrelid = ad.adrelid\n"
                + "WHERE\n"
                + "    n.nspname = :schemaName\n"
                + "    AND c.relname = :tableName\n"
                + "    AND a.attnum > 0\n"
                + "ORDER BY\n"
                + "    a.attnum";
    }

    @Override
    protected MetaColumn buildColumn(MetaColumn column, ResultSet resultSet) throws SQLException {
        MetaColumn metaColumn = super.buildColumn(column, resultSet);

        // dealingSpecialNumeric
        String typeName = metaColumn.getTypeName();
        Long precision = metaColumn.getPrecision();
        if (typeName.equals("numeric") && precision < 1) {
            metaColumn.setColumnType("numeric(38,10)");
            metaColumn.setPrecision(38L);
            metaColumn.setScale(10);
        }

        Object defaultValue = metaColumn.getDefaultValue();
        if (defaultValue != null && defaultValue.toString().contains("regclass")) {
            metaColumn.setDefaultValue(null);
        }
        return metaColumn;
    }

    @Override
    public String getCreateTableScript(BeeTableType type, String databaseName, String tableName) {
        try (Connection connection = connectionFactory.openConnection();
                Statement statement = connection.createStatement()) {

            if (BeeTableType.TABLE == type) {
                statement.execute(CREATE_TABLE_FUNCTION_SQL);
            }

            String tableSql = createTableSql(type, databaseName, tableName);
            ResultSet dbResult = statement.executeQuery(tableSql);
            if (dbResult.next()) {
                String ddl = dbResult.getString(getDDLColumnName(type));
                if (BeeTableType.VIEW == type) {
                    ddl = "CREATE VIEW " + databaseName + "." + tableName + " AS\n" + ddl;
                } else if (BeeTableType.MATERIALIZED_VIEW == type) {
                    ddl = "CREATE MATERIALIZED VIEW " + databaseName + "." + tableName + " AS\n" + ddl;
                }

                return ddl;
            } else {
                return "DDL is not available";
            }
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error", e);
        }
    }

    @Override
    protected BeeTableType getTableType(String type) {
        if (StringUtils.equalsIgnoreCase(type, "v")) {
            return BeeTableType.VIEW;
        } else if (StringUtils.equalsIgnoreCase(type, "r") || StringUtils.equalsIgnoreCase(type, "p")) {
            return BeeTableType.TABLE;
        } else if (StringUtils.equalsIgnoreCase(type, "m")) {
            return BeeTableType.MATERIALIZED_VIEW;
        } else {
            throw new IllegalArgumentException("table type is invalid: " + type);
        }
    }

    @Override
    protected String createTableSql(BeeTableType type, String schemaName, String tableName) {
        if (BeeTableType.TABLE == type) {
            return "select generate_create_table_statement('" + schemaName + "', '" + tableName + "') as ct_script;";
        } else {
            return "SELECT pg_get_viewdef('" + schemaName + "." + tableName + "') as ct_script";
        }
    }

    private static final String CREATE_TABLE_FUNCTION_SQL =
            "CREATE OR REPLACE FUNCTION generate_create_table_statement(p_schema_name varchar, p_table_name varchar)\n"
                    + "  RETURNS text AS\n"
                    + "$BODY$\n"
                    + "DECLARE\n"
                    + "    v_table_ddl   text;\n"
                    + "    column_record record;\n"
                    + "BEGIN\n"
                    + "    FOR column_record IN \n"
                    + "        SELECT \n"
                    + "            b.nspname as schema_name,\n"
                    + "            b.relname as table_name,\n"
                    + "            a.attname as column_name,\n"
                    + "            pg_catalog.format_type(a.atttypid, a.atttypmod) as column_type,\n"
                    + "            CASE WHEN \n"
                    + "                (SELECT substring(pg_catalog.pg_get_expr(d.adbin, d.adrelid) for 128)\n"
                    + "                 FROM pg_catalog.pg_attrdef d\n"
                    + "                 WHERE d.adrelid = a.attrelid AND d.adnum = a.attnum AND a.atthasdef) IS NOT NULL THEN\n"
                    + "                'DEFAULT '|| (SELECT substring(pg_catalog.pg_get_expr(d.adbin, d.adrelid) for 128)\n"
                    + "                              FROM pg_catalog.pg_attrdef d\n"
                    + "                              WHERE d.adrelid = a.attrelid AND d.adnum = a.attnum AND a.atthasdef)\n"
                    + "            ELSE\n"
                    + "                ''\n"
                    + "            END as column_default_value,\n"
                    + "            CASE WHEN a.attnotnull = true THEN \n"
                    + "                'NOT NULL'\n"
                    + "            ELSE\n"
                    + "                'NULL'\n"
                    + "            END as column_not_null,\n"
                    + "            a.attnum as attnum,\n"
                    + "            e.max_attnum as max_attnum\n"
                    + "        FROM \n"
                    + "            pg_catalog.pg_attribute a\n"
                    + "            INNER JOIN \n"
                    + "             (SELECT c.oid,\n"
                    + "                n.nspname,\n"
                    + "                c.relname\n"
                    + "              FROM pg_catalog.pg_class c\n"
                    + "                   LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace\n"
                    + "              WHERE c.relname ~ ('^('||p_table_name||')$') and n.nspname ~ ('^('||p_schema_name||')$')\n"
                    +
                    // "                AND pg_catalog.pg_table_is_visible(c.oid)\n" +
                    "              ORDER BY 2, 3) b\n"
                    + "            ON a.attrelid = b.oid\n"
                    + "            INNER JOIN \n"
                    + "             (SELECT \n"
                    + "                  a.attrelid,\n"
                    + "                  max(a.attnum) as max_attnum\n"
                    + "              FROM pg_catalog.pg_attribute a\n"
                    + "              WHERE a.attnum > 0 \n"
                    + "                AND NOT a.attisdropped\n"
                    + "              GROUP BY a.attrelid) e\n"
                    + "            ON a.attrelid=e.attrelid\n"
                    + "        WHERE a.attnum > 0 \n"
                    + "          AND NOT a.attisdropped\n"
                    + "        ORDER BY a.attnum\n"
                    + "    LOOP\n"
                    + "        IF column_record.attnum = 1 THEN\n"
                    + "            v_table_ddl:='CREATE TABLE \"'||column_record.schema_name||'\".\"'||column_record.table_name||'\" (';\n"
                    + "        ELSE\n"
                    + "            v_table_ddl:=v_table_ddl||',';\n"
                    + "        END IF;\n"
                    + " \n"
                    + "        IF column_record.attnum <= column_record.max_attnum THEN\n"
                    + "            v_table_ddl:=v_table_ddl||chr(10)||\n"
                    + "                     '    \"'||column_record.column_name||'\" '||column_record.column_type||' '||column_record.column_default_value||' '||column_record.column_not_null;\n"
                    + "        END IF;\n"
                    + "    END LOOP;\n"
                    + " \n"
                    + "    v_table_ddl:=v_table_ddl||');';\n"
                    + "    RETURN v_table_ddl;\n"
                    + "END;\n"
                    + "$BODY$\n"
                    + "  LANGUAGE 'plpgsql' COST 100.0 SECURITY INVOKER;";
}
